/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store.ha;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.store.CommitLog;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


// 记录着MQ服务器下用于master-slave服务器对象主从同步对象
public class HAService {
	
    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    // 记录着HAService对象中HAConnection对象计数器
    private final AtomicInteger connectionCount = new AtomicInteger(0);

    // 记录着HAService对象中HAConnection集合对象
    private final List<HAConnection> connectionList = new LinkedList<>();

    // 记录着HAService对象中对用于接收slave服务器连接请求对象
    private final AcceptSocketService acceptSocketService;

    // 记录着Broker服务器中用于存储Message消息的对象
    private final DefaultMessageStore defaultMessageStore;

    // 记录着HAService对象中用于阻塞线程对象实现同步|异步
    private final WaitNotifyObject waitNotifyObject = new WaitNotifyObject();
    
    // 记录着HAService对象中已确认同步成功CommitLog对象数据偏移量
    private final AtomicLong push2SlaveMaxOffset = new AtomicLong(0);

    // 记录着HAService对象中用于同步|异步等待对象
    private final GroupTransferService groupTransferService;

    // 记录着HAService对象中用于slave服务器与master服务器read|write对象
    private final HAClient haClient;

    // 创建HAService对象
    public HAService(final DefaultMessageStore defaultMessageStore) throws IOException {
        
    	this.defaultMessageStore = defaultMessageStore;
        // 创建accept服务用于接收slave服务器连接请求对象, default = 10912;
        this.acceptSocketService =
            new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
        
        this.groupTransferService = new GroupTransferService();
        this.haClient = new HAClient();
    }

    // 更新master服务器地址
    public void updateMasterAddress(final String newAddr) {
        if (this.haClient != null) {
            this.haClient.updateMasterAddress(newAddr);
        }
    }

    // 将GroupCommitRequest对象写入至GroupTransferService对象中
    public void putRequest(final CommitLog.GroupCommitRequest request) {
        this.groupTransferService.putRequest(request);
    }

    // 计算master服务器与slave服务器CommitLog数据偏移量之差
    public boolean isSlaveOK(final long masterPutWhere) {
        boolean result = this.connectionCount.get() > 0;
        result =
            result
                && ((masterPutWhere - this.push2SlaveMaxOffset.get()) < this.defaultMessageStore
                .getMessageStoreConfig().getHaSlaveFallbehindMax());
        // true : < 256M; false : >= 256M
        return result;
    }

    // 唤醒因主从同步而等待阻塞的线程对象
    public void notifyTransferSome(final long offset) {
        for (long value = this.push2SlaveMaxOffset.get(); offset > value; ) {
            boolean ok = this.push2SlaveMaxOffset.compareAndSet(value, offset);
            if (ok) {
                this.groupTransferService.notifyTransferSome();
                break;
            } else {
                value = this.push2SlaveMaxOffset.get();
            }
        }
    }

    // 获取HAConnection对象个数
    public AtomicInteger getConnectionCount() {
        return connectionCount;
    }

    // 启动Broker服务器下主从同步数据对象
    public void start() throws Exception {
    	// 创建ServerSocketChannel对象, 接收slave服务器连接请求
        this.acceptSocketService.beginAccept(); 
        
        this.acceptSocketService.start();
        this.groupTransferService.start();
        // 启动slave服务器对象
        this.haClient.start();
    }

    // 将HAConnection对象添加进connectionList对象中
    public void addConnection(final HAConnection conn) {
        synchronized (this.connectionList) {
            this.connectionList.add(conn);
        }
    }

    // 从connectionList中删除HAConnection对象
    public void removeConnection(final HAConnection conn) {
        synchronized (this.connectionList) {
            this.connectionList.remove(conn);
        }
    }

    // shutdown服务
    public void shutdown() {
        this.haClient.shutdown();
        this.acceptSocketService.shutdown(true);
        this.destroyConnections();
        this.groupTransferService.shutdown();
    }

    // 将master服务器与slave服务器对象下的HAConnection对象close
    public void destroyConnections() {
        synchronized (this.connectionList) {
            for (HAConnection c : this.connectionList) {
                c.shutdown();
            }

            this.connectionList.clear();
        }
    }

    // 获取DefaultMessageStore对象
    public DefaultMessageStore getDefaultMessageStore() {
        return defaultMessageStore;
    }

    public WaitNotifyObject getWaitNotifyObject() {
        return waitNotifyObject;
    }

    // 获取slave服务器已确认收到的CommitLog偏移量
    public AtomicLong getPush2SlaveMaxOffset() {
        return push2SlaveMaxOffset;
    }

    /**
     * Listens to slave connections to create {@link HAConnection}.
     */
    // 记录着master服务器用来接收slave服务器的连接请求对象
    class AcceptSocketService extends ServiceThread {
    	// 记录着master服务器下监听端口对象
        private final SocketAddress socketAddressListen;
        // 记录着master服务器下用于监听slave服务器连接的SocketChannel对象
        private ServerSocketChannel serverSocketChannel;
        
        // 记录着reactor模式轮询器
        private Selector selector;

        // 创建AcceptSocketService对象   (default = 10912)
        public AcceptSocketService(final int port) {
            this.socketAddressListen = new InetSocketAddress(port);
        }

        /**
         * Starts listening to slave connections.
         *
         * @throws Exception If fails.
         */
        // 设置ServerSokcetChannel对象并绑定端口对象, 注册OP_ACCEPT事件
        public void beginAccept() throws Exception {
            this.serverSocketChannel = ServerSocketChannel.open();
            // 创建Selector对象
            this.selector = RemotingUtil.openSelector();
            this.serverSocketChannel.socket().setReuseAddress(true);
            this.serverSocketChannel.socket().bind(this.socketAddressListen);
            this.serverSocketChannel.configureBlocking(false);
            this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
        }

        /**
         * {@inheritDoc}
         */
        @Override // shutdown AcceptSocketService线程对象
        public void shutdown(final boolean interrupt) {
            super.shutdown(interrupt);
            try { 
            	// close ServerSocketChannel对象
                this.serverSocketChannel.close();
                this.selector.close();
            } catch (IOException e) {
                log.error("AcceptSocketService shutdown exception", e);
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override// 启动AcceptSocketService线程对象, 用于接收slave服务器连接请求对象
        public void run() {
            log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                	// 阻塞线程
                    this.selector.select(1000);
                    // selected.size() <= 1
                    Set<SelectionKey> selected = this.selector.selectedKeys();

                    if (selected != null) {
                        for (SelectionKey k : selected) {
                        	// 判断网络操作位是否为OP_ACCEPT标志(只允许OP_ACCEPT)
                            if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                            	// 创建Tcp连接
                                SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();

                                if (sc != null) {
                                	// 向日志中记录master服务器与slave服务器连接建立成功
                                    HAService.log.info("HAService receive new connection, "
                                        + sc.socket().getRemoteSocketAddress());

                                    try {
                                    	// 创建HAConnection对象
                                        HAConnection conn = new HAConnection(HAService.this, sc);
                                        conn.start();
                                        // 将连接存入connectionList对象中
                                        HAService.this.addConnection(conn);
                                    } catch (Exception e) {
                                        log.error("new HAConnection exception", e);
                                        sc.close();
                                    }
                                }
                            } else {
                                log.warn("Unexpected ops in select " + k.readyOps());
                            }
                        }

                        // clear SelectKey集合对象
                        selected.clear();
                    }
                } catch (Exception e) {
                	// 异常发生
                    log.error(this.getServiceName() + " service has exception.", e);
                }
            }

            // 记录服务处于shutdown状态下
            log.info(this.getServiceName() + " service end");
        }

        /**
         * {@inheritDoc}
         */
        @Override // accept服务名称
        public String getServiceName() {
            return AcceptSocketService.class.getSimpleName();
        }
    }

    /**
     * GroupTransferService Service
     */
    // 它是用于处理主从服务,主服务器向从服务器发送数据(commitLog文件集群一致性)
    class GroupTransferService extends ServiceThread {

        private final WaitNotifyObject notifyTransferObject = new WaitNotifyObject();
        
        // 记录着写GroupCommitRequest集合对象, 最终写入requestsRead集合对象中
        private volatile List<CommitLog.GroupCommitRequest> requestsWrite = new ArrayList<>();
        // 记录着读GroupCommitRequest集合对象
        private volatile List<CommitLog.GroupCommitRequest> requestsRead = new ArrayList<>();


        // 将GroupCommitRequest同步写入requestsWrite集合对象中
        public synchronized void putRequest(final CommitLog.GroupCommitRequest request) {
            synchronized (this.requestsWrite) {
                this.requestsWrite.add(request);
            }
            if (hasNotified.compareAndSet(false, true)) {
                waitPoint.countDown(); // notify
            }
        }

        // 唤醒因等待主从同步而阻塞的线程对象
        public void notifyTransferSome() {
            this.notifyTransferObject.wakeup();
        }

        // 将requestsWrite与requestsRead引用地址进行交换
        private void swapRequests() {
            List<CommitLog.GroupCommitRequest> tmp = this.requestsWrite;
            this.requestsWrite = this.requestsRead;
            this.requestsRead = tmp;
        }

        private void doWaitTransfer() {
            synchronized (this.requestsRead) {
                if (!this.requestsRead.isEmpty()) {
                    for (CommitLog.GroupCommitRequest req : this.requestsRead) {
                        
                    	// true : 记录着master向slave服务器同步该偏移量数据成功
                    	boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
                        
                        // maxWaitTime = 5s
                        for (int i = 0; !transferOK && i < 5; i++) {
                        	// 阻塞线程对象1s
                            this.notifyTransferObject.waitForRunning(1000);
                            // true : 记录着master向slave服务器同步该偏移量数据成功
                            transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
                        }

                        // true : master服务器与slave服务器同步数据超时
                        if (!transferOK) {
                            log.warn("transfer messsage to slave timeout, " + req.getNextOffset());
                        }

                        // 唤醒因等待主从同步而阻塞的线程对象
                        req.wakeupCustomer(transferOK);
                    }

                    this.requestsRead.clear();
                }
            }
        }

        public void run() {
            log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                try {
                	// 阻塞线程对象10ms
                    this.waitForRunning(10);
                    // 计算master与slave服务器同步CommitLog中偏移量数据是否成功
                    this.doWaitTransfer();
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            log.info(this.getServiceName() + " service end");
        }

        @Override // 将requestsWrite对象与requestsRead对象互换身份
        protected void onWaitEnd() {
            this.swapRequests();
        }

        @Override // 获取GroupTransferService服务名称
        public String getServiceName() {
            return GroupTransferService.class.getSimpleName();
        }
    }

    // 记录着slave服务器向master服务器同步CommitLog对象中数据对象
    class HAClient extends ServiceThread {
    	// MaxBufferSize = 4M;
        private static final int READ_MAX_BUFFER_SIZE = 1024 * 1024 * 4;
        
        // 记录着slave服务器中的master服务器地址
        private final AtomicReference<String> masterAddress = new AtomicReference<>();
        
        // 记录着保存slave已确认收到的MaxCommitLog偏移量数据的ByteBuffer对象
        private final ByteBuffer reportOffset = ByteBuffer.allocate(8);
        
        // 记录着master服务器与slave服务器之间的SocketChannel对象
        private SocketChannel socketChannel;
        
        // 记录着slave服务器下的reactor模式轮询器
        private Selector selector;
        
        // 记录着slave服务器下最近从SocketChannel对象读时间戳
        private long lastWriteTimestamp = System.currentTimeMillis();

        // 记录着slave服务器下已向master服务器发送的已确认收到MaxCommitLog偏移量数据
        private long currentReportedOffset = 0;
        
        // 记录着slave服务器下byteBufferRead对象中已处理偏移量
        private int dispatchPostion = 0;
        // 记录着slave服务器下用于保存从master读取的ByteBuffer数据对象
        private ByteBuffer byteBufferRead = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);
        private ByteBuffer byteBufferBackup = ByteBuffer.allocate(READ_MAX_BUFFER_SIZE);

        // 创建HAClient对象,并使用selector轮询器
        public HAClient() throws IOException {
            this.selector = RemotingUtil.openSelector();
        }

        
        // 更新master服务器地址对象
        public void updateMasterAddress(final String newAddr) {
            String currentAddr = this.masterAddress.get();
            if (currentAddr == null || !currentAddr.equals(newAddr)) {
                this.masterAddress.set(newAddr);
                log.info("update master address, OLD: " + currentAddr + " NEW: " + newAddr);
            }
        }

        // 判断slave服务器是否应该向master服务器发送心跳包数据对象
        private boolean isTimeToReportOffset() {
            long interval =
                HAService.this.defaultMessageStore.getSystemClock().now() - this.lastWriteTimestamp;
            // true > 5s; 
            boolean needHeart = interval > HAService.this.defaultMessageStore.getMessageStoreConfig()
                .getHaSendHeartbeatInterval();

            return needHeart;
        }

        // 向master服务器发送slave服务器已确认收到的MaxCommitLog偏移量
        private boolean reportSlaveMaxOffset(final long maxOffset) {
        	// 将long型数据写入到ByteBuffer对象中
        	this.reportOffset.position(0);
            this.reportOffset.limit(8);
            this.reportOffset.putLong(maxOffset);
            this.reportOffset.position(0);
            this.reportOffset.limit(8);

            // maxWriteTimes = 3;
            for (int i = 0; i < 3 && this.reportOffset.hasRemaining(); i++) {
                try {
                	// 将MaxCommitLog数据写入master服务器
                    this.socketChannel.write(this.reportOffset);
                } catch (IOException e) {
                    log.error(this.getServiceName()
                        + "reportSlaveMaxOffset this.socketChannel.write exception", e);
                    return false;
                }
            }

            return !this.reportOffset.hasRemaining();
        }

        // 更新byteBufferRead(ByteBuffer对象)以便用于从SocketChannel中读数据
        private void reallocateByteBuffer() {
            int remain = READ_MAX_BUFFER_SIZE - this.dispatchPostion;
            
            if (remain > 0) {
                this.byteBufferRead.position(this.dispatchPostion);

                this.byteBufferBackup.position(0);
                this.byteBufferBackup.limit(READ_MAX_BUFFER_SIZE);
                this.byteBufferBackup.put(this.byteBufferRead);
            }

            this.swapByteBuffer();

            this.byteBufferRead.position(remain);
            this.byteBufferRead.limit(READ_MAX_BUFFER_SIZE);
            this.dispatchPostion = 0;
        }

        private void swapByteBuffer() {
            ByteBuffer tmp = this.byteBufferRead;
            this.byteBufferRead = this.byteBufferBackup;
            this.byteBufferBackup = tmp;
        }

        // SocketChannel对象中读字节数组对象并写入CommitLog对象中
        private boolean processReadEvent() {
            int readSizeZeroTimes = 0;
            // true : position = limit;
            while (this.byteBufferRead.hasRemaining()) {
                try {
                    int readSize = this.socketChannel.read(this.byteBufferRead);
                    if (readSize > 0) {
                    	// 记录写入时间戳
                        lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
                        readSizeZeroTimes = 0;
                        boolean result = this.dispatchReadRequest();
                        // true : result = false;
                        if (!result) {
                            log.error("HAClient, dispatchReadRequest error");
                            return false;
                        }
                    } else if (readSize == 0) {
                    	// MaxRetryRead = 3;
                        if (++readSizeZeroTimes >= 3) {
                            break;
                        }
                    } else {
                    	// SocketChannel未读入数据
                        log.info("HAClient, processReadEvent read socket < 0");
                        return false;
                    }
                } catch (IOException e) {
                	// 异常对象(slave服务器中读取master服务器数据出现异常)
                    log.info("HAClient, processReadEvent read socket exception", e);
                    return false;
                }
            }

            return true;
        }

        private boolean dispatchReadRequest() {
            final int msgHeaderSize = 8 + 4; // phyoffset + size
            int readSocketPos = this.byteBufferRead.position();

            while (true) {
            	
                int diff = this.byteBufferRead.position() - this.dispatchPostion;
                // true > 12;
                if (diff >= msgHeaderSize) {
                	
                	// 获取master服务器上已确认收到slave服务器MaxCommitLog偏移量
                    long masterPhyOffset = this.byteBufferRead.getLong(this.dispatchPostion);
                    // 获取Message对象所对应的ByteBuffer数据对象字节数组长度
                    int bodySize = this.byteBufferRead.getInt(this.dispatchPostion + 8);

                    // 获取slave服务器中的commitLog对象当前最大物理偏移量
                    long slavePhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();

                    if (slavePhyOffset != 0) { // 判断是否为第一次向主服务器发送数据
                    	// true : slavePhyOffset != masterPhyOffset
                        if (slavePhyOffset != masterPhyOffset) {
                            log.error("master pushed offset not equal the max phy offset in slave, SLAVE: "
                                + slavePhyOffset + " MASTER: " + masterPhyOffset);
                            return false;
                        }
                    }

                    // true : master服务器向slave服务器同步MessageExt对象
                    if (diff >= (msgHeaderSize + bodySize)) {
                    	// 获得body数据(以byte[]字节数组方法的List<MessageExt>集合对象)
                        byte[] bodyData = new byte[bodySize];
                        this.byteBufferRead.position(this.dispatchPostion + msgHeaderSize);
                        this.byteBufferRead.get(bodyData);

                        // 向CommitLog对象中写入MessageExt对象下的byte字节数组对象
                        HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);

                        // 设置position属性
                        this.byteBufferRead.position(readSocketPos);
                        this.dispatchPostion += msgHeaderSize + bodySize;

                        // slave服务器向master服务器发送当前CommitLog对象中的最大偏移量数据
                        if (!reportSlaveMaxOffsetPlus()) {
                            return false;
                        }

                        continue;
                    }
                }

                // true : ByteBuffer (position = limit)
                if (!this.byteBufferRead.hasRemaining()) {
                    this.reallocateByteBuffer();
                }

                break;
            }

            return true;
        }

        // slave服务器向master服务器发送CommitLog对象中的最大偏移量数据
        private boolean reportSlaveMaxOffsetPlus() {
            boolean result = true;
            // 获取最大物理写偏移量
            long currentPhyOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
            
            if (currentPhyOffset > this.currentReportedOffset) {
                this.currentReportedOffset = currentPhyOffset;
                
                // 将commitLog中的最大物理偏移量传给网络对端
                result = this.reportSlaveMaxOffset(this.currentReportedOffset);
                if (!result) {
                	// close SocketChannel对象
                    this.closeMaster();
                    log.error("HAClient, reportSlaveMaxOffset error, " + this.currentReportedOffset);
                }
            }

            return result;
        }

        // slave服务器向master服务器发起tcp请求并创建SocketChannel对象
        private boolean connectMaster() throws IOException {
            if (null == socketChannel) {
                String addr = this.masterAddress.get();
                if (addr != null) {

                	// 获取master服务器下所对应的SocketAddress对象
                    SocketAddress socketAddress = RemotingUtil.string2SocketAddress(addr);
                    if (socketAddress != null) {
                    	
                    	// 向master服务器发起TCP请求并获取SocketChannel对象
                        this.socketChannel = RemotingUtil.connect(socketAddress);
                        if (this.socketChannel != null) {
                        	// 向Selector对象注册OP_READ事件
                            this.socketChannel.register(this.selector, SelectionKey.OP_READ);
                        }
                    }
                }
                
                // 记录slave服务器中CommitLog对象中最大物理偏移量
                this.currentReportedOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
                
                // 记录着slave服务器最近从SocketChannel对象读时间戳
                this.lastWriteTimestamp = System.currentTimeMillis();
            }
            return this.socketChannel != null;
        }

        private void closeMaster() {
            if (null != this.socketChannel) {
                try {
                	// 获取在Selector上的SelectionKey对象,并撤销它
                    SelectionKey sk = this.socketChannel.keyFor(this.selector);
                    if (sk != null) {
                        sk.cancel();
                    }

                    // close SocketChannel对象
                    this.socketChannel.close();

                    this.socketChannel = null;
                } catch (IOException e) {
                    log.warn("closeMaster exception. ", e);
                }

                this.lastWriteTimestamp = 0;
                this.dispatchPostion = 0;

                this.byteBufferBackup.position(0);
                this.byteBufferBackup.limit(READ_MAX_BUFFER_SIZE);

                this.byteBufferRead.position(0);
                this.byteBufferRead.limit(READ_MAX_BUFFER_SIZE);
            }
        }

        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");

            // true : stop = false;
            while (!this.isStopped()) {
                try {
                	// 发起TCP连接请求
                    if (this.connectMaster()) {

                    	// true : 向master服务器发送心跳包数据携带CommitLog对象中最大偏移量数据
                        if (this.isTimeToReportOffset()) {
                        	// slave服务器向master服务器发送CommitLog对象中最大物理偏移量数据
                            boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);
                            if (!result) {
                                this.closeMaster();
                            }
                        }

                        // 阻塞select
                        this.selector.select(1000);

                        // SocketChannel对象中读字节数组对象并写入CommitLog对象中
                        boolean ok = this.processReadEvent();
                        if (!ok) {
                            this.closeMaster();
                        }

                        // slave服务器向master服务器发送CommitLog对象中的最大偏移量数据
                        if (!reportSlaveMaxOffsetPlus()) {
                            continue;
                        }

                        long interval =
                            HAService.this.getDefaultMessageStore().getSystemClock().now()
                                - this.lastWriteTimestamp;
                        
                        // true : > 20s;
                        if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig()
                            .getHaHousekeepingInterval()) {
                            log.warn("HAClient, housekeeping, found this connection[" + this.masterAddress
                                + "] expired, " + interval);
                            this.closeMaster();
                            log.warn("HAClient, master not response some time, so close connection");
                        }
                    } else {
                    	// 阻塞指定时间(5s)
                        this.waitForRunning(1000 * 5);
                    }
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                    this.waitForRunning(1000 * 5);
                }
            }

            // 记录服务shutdown
            log.info(this.getServiceName() + " service end");
        }

        @Override// 当前服务名称
        public String getServiceName() {
            return HAClient.class.getSimpleName();
        }
    }
}
